package com.bc.commoncore.concurrent;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.bc.commoncore.util.StringUtils;

/**
 * 任务并发/并行工具类
 * @author jibiao.liu
 */
public class ConcurrentTask {

	private static final Logger LOGGER = LoggerFactory.getLogger(ConcurrentTask.class);
	
	private CountDownLatch countDown = null;
	/** 并行任务列表*/
	private List<Runnable> tasks = new ArrayList<Runnable>();
	/** 同步任务列表*/
	private List<SyncTaskObj> syncTasks = new ArrayList<SyncTaskObj>();
	/** Key集合列表,防止重复*/
	private List<String> keyList = new ArrayList<String>();
	/** 是否同步,默认并行*/
	private Boolean isSync = false;
	private int taskSize = 0;
	private long timeout = 0;
	private volatile Map<String, MsgResult> result = new ConcurrentHashMap<String, MsgResult>();
	
/**
* 并行、同步切换构造函数
 * @param taskSize
 * @param isSync
 * @param timeout 单位是秒
 */
	public ConcurrentTask(int taskSize, Boolean isSync, long timeout) {
		this.isSync = isSync;
		this.timeout = timeout;
		if (!isSync) {
			this.taskSize = taskSize;
			countDown = new CountDownLatch(taskSize);	
		}
	}
	/**
	 * 并行、同步切换构造函数
	 * @param taskSize 任务数量
	 * @param isSync 是否同步
	 */
	public ConcurrentTask(int taskSize, Boolean isSync) {
		this(taskSize,isSync,0);
	}
/**
 * 
 * @param taskSize
 * @param timeout 超时时间，单位是秒
 */
	public ConcurrentTask(int taskSize, long timeout) {
		this(taskSize,false,timeout);
	}
	public ConcurrentTask(int taskSize) {
		this(taskSize,false,0);
	}
	/**
	 * 同步构造函数
	 */
	public ConcurrentTask() {
		this.isSync = true;
	}
	
	/**
	 * 创建并行任务
	 * @param mapper
	 * @param param
	 * @param keyName
	 */
	private void createConcurrentTask(Function mapper, Object param, String keyName) {
		Runnable runnable = new Runnable() {
			@Override
			public void run() {
				long sTime = System.currentTimeMillis();
				try {
					Object obj = mapper.apply(param);
					MsgResult msg = new MsgResult(1, "", obj, System.currentTimeMillis() - sTime);
					synchronized (result) {
						result.put(keyName, msg);
					}
				} catch (Exception e) {
					MsgResult msg = new MsgResult(-1, "", StringUtils.getStackTrace(e), System.currentTimeMillis() - sTime);
					synchronized (result) {
						result.put(keyName, msg);
					}
				} finally {
					// 计数
					countDown.countDown();	
				}
			}
		};
		tasks.add(runnable);
	}
	
	/**
	 * 运行并行任务集合
	 * @param executorService
	 */
	private void runConcurrentTask(ExecutorService executorService) {
		// 执行任务
		for (Runnable task : tasks) {
			executorService.execute(task);
		}
	}
	
	// ----------------------------------------------------- 同步业务
	
	/**
	 * 创建同步任务
	 * @param mapper
	 * @param param
	 * @param keyName
	 */
	private void createSyncTask(Function mapper, Object param, String keyName) {
		SyncTaskObj sto = new SyncTaskObj();
		sto.setMapper(mapper);
		sto.setParam(param);
		sto.setKeyName(keyName);
		
		syncTasks.add(sto);
	}
	
	/**
	 * 运行同步任务集合
	 * @param executorService
	 */
	private void runSyncTask() {
		long sTime = System.currentTimeMillis();
		// 执行任务
		for (SyncTaskObj task : syncTasks) {
			Function mapper = task.getMapper();
			Object param = task.getParam();
			String keyName = task.getKeyName();
			
			Object obj = mapper.apply(param);
			MsgResult msg = new MsgResult(1, "", obj, System.currentTimeMillis() - sTime);
			synchronized (result) {
				result.put(keyName, msg);
			}
		}
	}
	
	// --------------------------------------------- 执行步骤
	
	/**
	 * 第一步:添加任务
	 * @param mapper
	 * @param param
	 * @param keyName
	 */
	public void addTask(Function mapper, Object param, String keyName) {
		if (keyList.contains(keyName)) {
			throw new RuntimeException("添加并行任务时异常:"+keyName+"已经存在!");
		}
		if (isSync) {
			createSyncTask(mapper, param, keyName);
		} else {
			createConcurrentTask(mapper, param, keyName);	
		}
		keyList.add(keyName);
	}
	
	/**
	 * 第二步:开始执行
	 */
	public void start() {
		if (isSync) {
			runSyncTask();
		} else {
			if (tasks.size() != taskSize) {
				throw new RuntimeException(tasks.size()+","+taskSize+"任务数量不一致异常，设置的任务数量与实际任务数不一致!");
			}
			taskSize = tasks.size();
			ExecutorService executorService = Executors
					.newFixedThreadPool(taskSize);
			runConcurrentTask(executorService);
			try {
				// 等待所有任务完成
				if (timeout>0) {
					boolean timeoutFlag = countDown.await(timeout, TimeUnit.SECONDS);
					if (timeoutFlag) {
						LOGGER.info("所有子线程执行完成");
					} else {
						throw new RuntimeException("并行任务超时("+timeout+"s)!");
					} 
				}else {
					countDown.await();
				}
			} catch (Exception e) {
				LOGGER.error("倒计数锁存器中断异常!", e);
			} finally {
				try {
					executorService.shutdownNow();
				} catch (Exception e) {
					LOGGER.error("线程池关闭异常!", e);
				}
			}
		}
	}
	
	public Map<String, MsgResult> getResults() {
		return result;
	}
}
